airbyte/_executors/util.py (1)

268-280: Bug: colon-based tag check breaks for images with a registry port (e.g., host:5000/repo/image).

The condition if version is not None and ":" in docker_image: will incorrectly treat images from registries with a port (e.g., my-registry:5000/airbyte/source-faker) as if they already include a tag. This will:

• Wrongly raise PyAirbyteInputError when the user supplies version along with a ported image without a tag.
• Bypass your new version resolution logic, implicitly relying on Docker’s default “latest” tag.

Could we detect tags/digests only in the last path segment (and consider @digest) to disambiguate ports, wdyt?

Proposed fix:

-        if version is not None and ":" in docker_image:
+        # Detect explicit tag/digest ignoring registry ports (e.g., ghcr.io:443/..., localhost:5000/...)
+        name_segment = docker_image.rsplit("/", 1)[-1]
+        has_tag = ":" in name_segment
+        has_digest = "@" in docker_image  # e.g., image@sha256:...
+        if version is not None and (has_tag or has_digest):
             raise exc.PyAirbyteInputError(
                 message="The 'version' parameter is not supported when a tag is already set in the "
                 "'docker_image' parameter.",
                 context={
                     "docker_image": docker_image,
                     "version": version,
                 },
             )

airbyte/_executors/util.py (3)

31-31: LGTM on module-level logger.

Using a per-module logger is the right approach for library code. Do you also want to consider adding a NullHandler at the package root so library users don’t see “No handler found” warnings if they haven’t configured logging, wdyt?

---

283-304: Optional: tailor the warning message when AIRBYTE_OFFLINE_MODE is enabled.

Right now the same “could not be determined from the registry” message is used both for offline mode and unknown metadata. Would it reduce confusion to explicitly mention “offline mode is enabled; skipping registry lookup” when AIRBYTE_OFFLINE_MODE is true, wdyt?

---

268-304: Add unit tests for edge cases (ports/digests/offline).

Given environment constraints, could we cover the decision logic with lightweight unit tests that mock registry and environment, wdyt? Suggested cases:

• docker_image="airbyte/source-faker" with no version: picks metadata.latest_available_version.
• AIRBYTE_OFFLINE_MODE=1: falls back to VERSION_LATEST and logs the offline warning.
• docker_image="my-registry:5000/airbyte/source-faker" with no version: still appends resolved version (port should not be treated as a tag).
• docker_image="airbyte/source-faker@sha256:…" with or without version: does not append version.
• docker_image="airbyte/source-faker:1.2.3" with version provided: raises PyAirbyteInputError.
• docker_image="my-registry:5000/airbyte/source-faker" with version="1.2.3": appends ":1.2.3".

I can draft pytest-style tests that monkeypatch get_connector_metadata, set AIRBYTE_OFFLINE_MODE, and assert the constructed docker_cmd args without invoking Docker. Want me to put together a test scaffold, wdyt?

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

airbyte/_executors/util.py (1)

272-296: Fix tag/digest detection (ports misclassified), use VERSION_LATEST, and resolve lint E501.

• The test if ":" not in docker_image misclassifies images from registries with ports (e.g. my-reg:5000/airbyte/source) and also doesn’t account for digests. This can skip your explicit tag resolution and undermine the PR’s goal.
• Reuse a single detection for has_tag and has_digest, and use it both where you validate version conflicts and where you decide to append a tag.
• Prefer VERSION_LATEST constant over the string literal "latest".
• Address the Ruff E501 failure at Line 292 by wrapping the warning string.
• Optional but clearer: tailor the warning for offline mode vs registry/metadata issues (matches prior review suggestions), wdyt?

Proposed patch:

@@
-        if version is not None and ":" in docker_image:
+        # Detect whether the image already includes an explicit tag or digest.
+        last_slash = docker_image.rfind("/")
+        has_digest = "@" in docker_image
+        has_tag = (":" in docker_image[last_slash + 1:]) and not has_digest
+
+        if version is not None and (has_tag or has_digest):
             raise exc.PyAirbyteInputError(
                 message="The 'version' parameter is not supported when a tag is already set in the "
                 "'docker_image' parameter.",
                 context={
                     "docker_image": docker_image,
                     "version": version,
                 },
             )
@@
-        if ":" not in docker_image:
+        if not (has_tag or has_digest):
             resolved_version = version
             if resolved_version is None:
                 if metadata and metadata.latest_available_version:
                     resolved_version = metadata.latest_available_version
                 else:
-                    resolved_version = "latest"
-                    if logger:
-                        logger.warning(
-                            f"Using 'latest' tag for connector '{name}' because no explicit "
-                            f"version was specified and we could not locate the latest version number."
-                        )
+                    resolved_version = VERSION_LATEST
+                    # Lazily get a logger to avoid import-time side effects.
+                    local_logger = logger or get_global_file_logger()
+                    if local_logger:
+                        # Provide a more specific warning depending on the condition and keep lines <= 100 chars.
+                        if AIRBYTE_OFFLINE_MODE:
+                            msg = (
+                                f"Using '{VERSION_LATEST}' tag for connector '{name}' because PyAirbyte "
+                                "is running in offline mode and no explicit version was specified."
+                            )
+                        elif metadata is None:
+                            msg = (
+                                f"Using '{VERSION_LATEST}' tag for connector '{name}' because the registry "
+                                "could not be reached and no explicit version was specified."
+                            )
+                        else:
+                            msg = (
+                                f"Using '{VERSION_LATEST}' tag for connector '{name}' because no explicit "
+                                "version was specified and no latest version was found in the registry."
+                            )
+                        local_logger.warning(msg)
 
             docker_image = f"{docker_image}:{resolved_version}"

This also resolves the pipeline error Ruff E501 line too long (Line 292) by wrapping the message. Wdyt?

airbyte/_executors/util.py (1)

31-31: Consider lazy-initializing the logger to avoid import-time side effects, wdyt?

get_global_file_logger writes to disk and prints a message to stderr when configured. Doing this at import time can introduce unintended side effects. Would you consider lazy-initializing at the first log site instead? You’re already guarding with an existence check, so a localized lazy lookup is straightforward.

Example inside the warning site (no global reassignment needed):

-                    if logger:
-                        logger.warning(
+                    local_logger = logger or get_global_file_logger()
+                    if local_logger:
+                        local_logger.warning(
                             f"Using 'latest' tag for connector '{name}' because no explicit "
                             f"version was specified and we could not locate the latest version number."
                         )

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

airbyte/_executors/util.py (2)

272-297: Bug: tag detection misclassifies images with registry ports and digest refs; also prefer VERSION_LATEST and clearer warning reasons.

• ":" in docker_image treats registry ports (e.g., ghcr.io:443/airbyte/source) as if a tag is set, skipping explicit version resolution. That undermines the PR’s goal and can accidentally use Docker’s implicit latest without warning.
• Digest refs (image@sha256:...) should be treated as “explicit reference present” and skip adding a tag; current logic doesn’t account for that.
• Use VERSION_LATEST instead of the literal "latest".
• Split warnings for offline mode vs metadata unavailability for clarity (as suggested earlier).

Proposed fix (includes robust reference parsing and clearer warnings):

@@
-        if version is not None and ":" in docker_image:
+        # Determine whether the docker reference already has an explicit tag or digest.
+        has_tag, has_digest = _docker_ref_has_explicit_reference(docker_image)
+        if version is not None and (has_tag or has_digest):
             raise exc.PyAirbyteInputError(
                 message="The 'version' parameter is not supported when a tag is already set in the "
                 "'docker_image' parameter.",
                 context={
                     "docker_image": docker_image,
                     "version": version,
                 },
             )
@@
-        if ":" not in docker_image:
+        if not (has_tag or has_digest):
             resolved_version = version
             if resolved_version is None:
-                if metadata and metadata.latest_available_version:
+                if metadata and getattr(metadata, "latest_available_version", None):
                     resolved_version = metadata.latest_available_version
                 else:
-                    resolved_version = "latest"
-                    if logger:
-                        logger.warning(
-                            f"Using 'latest' tag for connector '{name}' because no explicit "
-                            f"version was specified and we could not locate the latest "
-                            f"version number."
-                        )
+                    resolved_version = VERSION_LATEST
+                    if logger:
+                        if AIRBYTE_OFFLINE_MODE:
+                            logger.warning(
+                                f"Using '{VERSION_LATEST}' tag for connector '{name}' because no explicit "
+                                f"version was specified and registry lookups are disabled in offline mode. "
+                                f"Consider specifying an explicit version or disabling offline mode to fetch "
+                                f"the latest version from the registry."
+                            )
+                        elif metadata is None:
+                            logger.warning(
+                                f"Using '{VERSION_LATEST}' tag for connector '{name}' because no explicit "
+                                f"version was specified and connector metadata is unavailable. "
+                                f"Consider specifying an explicit version or check your internet connection."
+                            )
+                        else:
+                            logger.warning(
+                                f"Using '{VERSION_LATEST}' tag for connector '{name}' because no explicit "
+                                f"version was specified and the registry did not provide a latest_available_version."
+                            )
 
             docker_image = f"{docker_image}:{resolved_version}"

Helper to add (outside this range), to robustly detect tags/digests (colon after last slash means a tag; “@” means digest):

def _docker_ref_has_explicit_reference(image: str) -> tuple[bool, bool]:
    """Return (has_tag, has_digest) for a docker image reference.

    - Treat colon after the last slash as a tag (e.g., repo/name:1.2.3).
    - Treat '@' as digest (e.g., repo/name@sha256:...).
    - Avoid misclassifying registry ports (e.g., ghcr.io:443/repo/name).
    """
    if "@" in image:
        return False, True
    last_slash = image.rfind("/")
    last_colon = image.rfind(":")
    has_tag = last_colon > last_slash  # colon after last slash => tag, not registry port
    return has_tag, False

This change ensures we only append a tag when neither a tag nor a digest is present, while still allowing registries with ports. It also clarifies the fallback reason for users and uses the VERSION_LATEST constant. Wdyt?

---

288-294: Use the VERSION_LATEST constant instead of the string literal and tailor the warning.

Minor nit: please switch "latest" to VERSION_LATEST and consider distinguishing offline mode vs missing metadata to improve operator clarity, wdyt?

airbyte/_executors/util.py (1)

31-31: Consider lazy/fallback logging to avoid silent warnings when logger is None.

get_global_file_logger() can return None; the current code no-ops in that case. Do we want a minimal stderr fallback to ensure users see the warning when a file logger isn’t configured, wdyt?

For example, you could centralize warn emission:

def _warn(msg: str) -> None:
    if logger:
        logger.warning(msg)
    else:
        print(f"[warn] {msg}", file=sys.stderr)

…and then call _warn(...) below.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro